# std
import os
import datetime
import functools
import collections

# 3rd party
import pandas as pd
import numpy
import ujson
from absl import app, flags

# internal
from coin.proto.coin_order_gateway_pb2 import OrderEvent, OrderGatewayLog
import coin.proto.coin_order_enums_pb2 as coin_order
from coin.proto.support.coin_order_stateless_pb2 import StatelessOrderEvent, OrderStateTrace
from coin.support.proto_log.logic.util import run_from_og_log_archive
from xunkemgmt_client.client.api_client import XunkemgmtClient
from coin2.service.strategy import trading_summary_service_pb2 as tss_pb2
from coin.base.datetime_util import to_timestamp_int

import coin.strategy.mm.tool.archive_base as abase
from coin.pnl.sim_stat_calc2 import bisect_batch
from coin.exchange.kr_rest.product.product_impl import generate_product_from_str2

FEED_MACHINES = [
    'feed-05.cn-hongkong.aliyun',
    'feed-05.ap-northeast-1.aws',
    'feed-01.ap-northeast-2.aws',
    'feed-01.eu-west-1.aws',
    'feed-01.us-east-1.aws',
]

tws = ['m10000', 'm3000', 'm1000', 0, 25, 50, 100, 200, 500, 1000, 3000, 5000, 10000, 30000, 60000]

Record = collections.namedtuple(
    'Record',
    [
        'api_version',
        'order_machine',
        'feed_machine',
        'market_type',
        'exchange',
        'symbol',
        'timestamp',
        'sign',
        'proc_order_id',
        'spread_bps',
    ] + ['order_ret_bps_%sms_fwd' % time for time in tws]
    + ['payup0_ret_bps_%sms_fwd' % time for time in tws]
    + ['mid_ret_bps_%sms_fwd' % time for time in tws]
    + ['order_payup_bps_%sms_fwd' % time for time in tws]
    + ['fill_ret_bps_%sms_fwd' % time for time in tws]
    + ['order_qty', 'fill_qty', 'taker_fill_qty', 'lifetime_ms', 'signal', 'model_id'])

ORDER_ROOT_DIR = '/remote/iosg/coin/data/og_proto_stateless_log'
FEED_ROOT_DIR = None
FEED_CACHE_DIR = '/remote/iosg/coin-2/feed_cache'

import getpass

OUT_DIR = f'/remote/iosg/home-2/{getpass.getuser()}/order_latency'

flags.DEFINE_string('date', None, 'example: --date 20190708')
flags.DEFINE_string('strat_group', None, 'example: --strat_group lm-agg')
flags.DEFINE_string('strat_name', None, 'example: --strat_name huobi_futures_quarter_self')
flags.DEFINE_string('symbol', None, 'example: BTC')
flags.DEFINE_string('prefix', '', '')

Bbo = collections.namedtuple('Bbo', ('timestamp', 'price', 'qty'))

# bbo history for one symbol


class ProductBboInfo(object):
  def __init__(self):
    super().__init__()
    self.bbo_asks = []
    self.bbo_bids = []

  def update(self, book_event):
    book = book_event
    askp = book.askp
    bidp = book.bidp

    if askp:
      bbo_ask = Bbo(timestamp=book.timestamp, price=askp, qty=askp)
      # price diff only
      if not self.bbo_asks or self.bbo_asks[-1][1] != bbo_ask[1]:
        self.bbo_asks.append(bbo_ask)

    if bidp:
      bbo_bid = Bbo(timestamp=book.timestamp, price=bidp, qty=bidp)
      # price diff only
      if not self.bbo_bids or self.bbo_bids[-1][1] != bbo_bid[1]:
        self.bbo_bids.append(bbo_bid)


class StatelessConvert(object):
  def __init__(self, symbol, start_ts, end_ts):
    self.trace = collections.defaultdict(lambda: OrderStateTrace())
    self.orders = []
    self.start_ts = start_ts
    self.end_ts = end_ts
    self.symbol = symbol

  def on_log(self, timestamp, log):
    if timestamp < self.start_ts or timestamp >= self.end_ts:
      return

    pb = OrderGatewayLog()
    try:
      pb.ParseFromString(log)
    except Exception:
      print(log)
      return

    if pb.type != OrderGatewayLog.ORDER_EVENT:
      return

    ev = pb.event

    if ev.symbol != self.symbol:
      return

    if ev.event_time == 0:
      raise RuntimeError('event_time == 0')

    ioid = ev.internal_order_id or ev.proc_order_id

    # order_state
    if ev.order_state == coin_order.CREATED_ORDER:
      if self.trace[ioid].created_time == 0:
        self.trace[ioid].created_time = ev.event_time
    elif ev.order_state == coin_order.PENDING_ORDER:
      if self.trace[ioid].pending_time == 0:
        self.trace[ioid].pending_time = ev.event_time
    elif ev.order_state == coin_order.WORKING_ORDER:
      if self.trace[ioid].working_time == 0:
        self.trace[ioid].working_time = ev.event_time
    elif ev.order_state == coin_order.DEAD_ORDER:
      if self.trace[ioid].dead_time == 0:
        self.trace[ioid].dead_time = ev.event_time

    # order_event for order_state
    if ev.type == OrderEvent.ORDER_SUBMITTED:
      if ev.tag != "" and ev.tag[0] == '{':
        tag_obj = ujson.loads(ev.tag)
        self.trace[ioid].signal = tag_obj.get('signal', None)
        self.trace[ioid].model_id = tag_obj['model_id']
      if self.trace[ioid].pending_time == 0:
        self.trace[ioid].pending_time = ev.event_time
    elif ev.type == OrderEvent.ORDER_ACCEPTED:
      if self.trace[ioid].working_time == 0:
        self.trace[ioid].working_time = ev.event_time
    elif ev.type in [
        OrderEvent.ORDER_EXPIRED,
        OrderEvent.ORDER_REJECTED,
        OrderEvent.CANCEL_CONFIRMED,
        OrderEvent.ORDER_AUTO_CANCELED
    ]:
      if self.trace[ioid].dead_time == 0:
        self.trace[ioid].dead_time = ev.event_time

    # order_event
    if ev.type == OrderEvent.UNKNOWN_ORDER_EVENT:
      if self.trace[ioid].unknown_order_event_time == 0:
        self.trace[ioid].unknown_order_event_time = ev.event_time
    elif ev.type == OrderEvent.ORDER_SUBMITTED:
      if self.trace[ioid].order_submitted_time == 0:
        self.trace[ioid].order_submitted_time = ev.event_time
    elif ev.type == OrderEvent.ORDER_ACCEPTED:
      if self.trace[ioid].order_accepted_time == 0:
        self.trace[ioid].order_accepted_time = ev.event_time
    elif ev.type == OrderEvent.ORDER_ERROR:
      if self.trace[ioid].order_error_time == 0:
        self.trace[ioid].order_error_time = ev.event_time
    elif ev.type == OrderEvent.ORDER_FILLED:  # special case
      lst = self.trace[ioid].order_filled.add()
      lst.timestamp = ev.event_time
      lst.qty = ev.fill_qty
      lst.price = ev.fill_price
      lst.taker_qty = ev.fill_qty * ((ev.fill_type == coin_order.TAKER_FILL_TYPE) |
                                     (ev.guess_fill_type == coin_order.TAKER_FILL_TYPE))
    elif ev.type == OrderEvent.ORDER_AUTO_CANCELED:
      if self.trace[ioid].order_auto_canceled_time == 0:
        self.trace[ioid].order_auto_canceled_time = ev.event_time
    elif ev.type == OrderEvent.ORDER_AMENDED:
      if self.trace[ioid].order_amended_time == 0:
        self.trace[ioid].order_amended_time = ev.event_time
    elif ev.type == OrderEvent.ORDER_AMEND_ACCEPTED:
      if self.trace[ioid].order_amend_accepted_time == 0:
        self.trace[ioid].order_amend_accepted_time = ev.event_time
    elif ev.type == OrderEvent.ORDER_REJECTED:
      if self.trace[ioid].order_rejected_time == 0:
        self.trace[ioid].order_rejected_time = ev.event_time
    elif ev.type == OrderEvent.ORDER_EXPIRED:
      if self.trace[ioid].order_expired_time == 0:
        self.trace[ioid].order_expired_time = ev.event_time
    elif ev.type == OrderEvent.CANCEL_SUBMITTED:
      if self.trace[ioid].cancel_submitted_time == 0:
        self.trace[ioid].cancel_submitted_time = ev.event_time
    elif ev.type == OrderEvent.CANCEL_ACCEPTED:
      if self.trace[ioid].cancel_accepted_time == 0:
        self.trace[ioid].cancel_accepted_time = ev.event_time
    elif ev.type == OrderEvent.CANCEL_ERROR:
      if self.trace[ioid].cancel_error_time == 0:
        self.trace[ioid].cancel_error_time = ev.event_time
    elif ev.type == OrderEvent.CANCEL_CONFIRMED:
      if self.trace[ioid].cancel_confirmed_time == 0:
        self.trace[ioid].cancel_confirmed_time = ev.event_time

    out = StatelessOrderEvent()
    out.event.CopyFrom(ev)
    out.state.CopyFrom(self.trace[ioid])

    if pb.event.type in [OrderEvent.ORDER_FILLED, OrderEvent.CANCEL_CONFIRMED]:
      self.orders.append(out)


def get_order_fills_cancels(market_type, exchange, date, machine, owner, symbol):

  start_time = datetime.datetime.strptime(date, '%Y%m%d').replace(tzinfo=datetime.timezone.utc)
  end_time = start_time + datetime.timedelta(days=1)

  start_ts = int(
      datetime.datetime.timestamp(start_time.replace(tzinfo=datetime.timezone.utc)) * 1e9)
  end_ts = int(datetime.datetime.timestamp(end_time.replace(tzinfo=datetime.timezone.utc)) * 1e9)

  sconv = StatelessConvert(symbol, start_ts, end_ts)

  run_from_og_log_archive(sconv.on_log,
                          market_type=market_type,
                          exchange=exchange,
                          machine=machine,
                          owner=owner,
                          root_dir='/remote/iosg/strat-1/buckets/log.raw.coin/live/og_proto_log/',
                          start_time=start_time,
                          end_time=end_time)

  return sconv.orders


BookEvent = collections.namedtuple('BookEvent', ['product', 'timestamp', 'bidp', 'askp'])


def get_feed_stream(date, products, machine) -> pd.DataFrame:
  start_time = datetime.datetime.strptime(date, '%Y%m%d')
  end_time = start_time + \
      datetime.timedelta(days=1) + datetime.timedelta(minutes=35 + 35)

  feeds = []

  def on_book(product, book):
    if book is None:
      return
    if book.bid0() is None or book.ask0() is None:
      return
    feeds.append(BookEvent(product, book.timestamp, book.bid0().price, book.ask0().price))

  def on_trade(product, trade):
    # feeds.append((product, trade.timestamp, 'trade', trade))
    pass

  def on_book_reset(_book_builder_name, book_builder):
    for product in products:
      book_builder.subscribe(product, functools.partial(on_book, product))
      if hasattr(book_builder, "subscribe_trade"):
        book_builder.subscribe_trade(product, functools.partial(on_trade, product))

  coin1_feed_api = abase.get_api_override_from_override_string(
      "Spot:Okex=v3,Futures:Okex=v3,Spot:Binance=l1_realtime,"
      + "Futures:Binance=l1_realtime,Spot:Quoinex=single_channel")

  coin2_feed_api = abase.get_api_override_from_override_string(
      "Spot:Okex=v3.realtime_light_bbo,Futures:Okex=v3.realtime_light_bbo,"
      + "Spot:Binance=v1.l1_realtime,Futures:Binance=v1.realtime_light_bbo,"
      + "Spot:Huobi=v1.snapshot,Futures:Huobi=v1.realtime_light_bbo,"
      + "Futures:Bitmex=v1.realtime_light_bbo")
  if machine.startswith("feed-05") >= 0:
    api_override = coin2_feed_api
  else:
    api_override = coin1_feed_api
  print(api_override)

  print(products)
  abase.run_from_feed_cache_base(products,
                                 on_book_reset,
                                 start_time,
                                 end_time,
                                 machine,
                                 worker_ids=['1', '2'],
                                 write_feed_cache_if_none=False,
                                 feed_cache_dir=FEED_CACHE_DIR,
                                 api_override=api_override)
  return feeds


# HACK, copied from archive_base.py
# this script does not use baseflags
def order_symbols_to_products(symbols, date, market_type, exchange):
  # .replace(tzinfo=datetime.timezone.utc)
  start_time = datetime.datetime.strptime(date, '%Y%m%d')
  products = []
  for symbol in symbols:
    product = generate_product_from_str2(market_type=market_type,
                                         exchange=exchange,
                                         api_version=None,
                                         norm_product=symbol,
                                         current_datetime=start_time)
    products.append(product)
  return products


def region_from_machine(m):
  return m.split('.', 3)[1]


def get_rows_records(rows, date):
  recorddfs = []
  bbos = {}
  for row in rows:
    print(str(pd.DataFrame(row).T) + "\n")
    strategy_region = region_from_machine(row.machine)
    if "us-east-2" in strategy_region:
      strategy_region = strategy_region.replace("us-east-2", "us-east-1")

    if strategy_region == 'aliyun-cn-hongkong':
      strategy_region = 'cn-hongkong'
    feed_machines = list(filter(lambda m: region_from_machine(m) == strategy_region, FEED_MACHINES))

    if not feed_machines:
      # e.g. Kraken
      print('no feed machine for strategy-region', strategy_region)
      return []

    # list of StatelessOrderEvent
    orders = get_order_fills_cancels(row.market_type,
                                     row.exchange,
                                     date,
                                     row.machine,
                                     row.owner,
                                     row.symbol)
    print(f'{len(orders)} orders')
    if len(orders) == 0:
      continue

    symbols = [row.symbol]

    products = order_symbols_to_products(symbols, date, row.market_type, row.exchange)
    order_symbol_to_product_map = {symbol: product for (symbol, product) in zip(symbols, products)}

    assert len(feed_machines) == 1, f'more than feed machines: {feed_machines}'
    feed_machine = feed_machines[0]
    # book + trade => book only

    for product in products:
      if product not in bbos:
        bbos[product] = ProductBboInfo()
        feeds = get_feed_stream(date, [product], feed_machine)
        print(f'{len(feeds)} feeds: {row.market_type} {row.exchange} {product}')
        for feed in feeds:
          bbos[product].update(feed)

    # trace = {}
    # order_by_oid = {}

    from google.protobuf.json_format import MessageToDict

    arows = []
    for order in orders:
      adict = MessageToDict(order.event, preserving_proto_field_name=True)
      adict.update(MessageToDict(order.state, preserving_proto_field_name=True))

      is_taker = order.event.fill_type == coin_order.TAKER_FILL_TYPE
      adict['taker_fill_qty'] = order.event.fill_qty * is_taker
      arows.append(adict)

    df = pd.DataFrame(arows)
    df['fill_pq'] = df['fill_price'] * df['fill_qty']

    sumdf = df.groupby('proc_order_id').sum()[[
        'fill_qty', 'taker_fill_qty', 'fill_pq',
        'cancelled_qty']]
    columns = [
        'order_submitted_time', 'symbol', 'market_type', 'order_side',
        'dead_time', 'order_price', 'order_qty', 'post_only']
    if 'signal' in df.columns:
      columns.append('signal')
    lastdf = df.groupby('proc_order_id').last()[columns]

    lastdf['fill_price'] = sumdf['fill_pq'] / sumdf['fill_qty']
    lastdf['fill_price'] = lastdf['fill_price'].fillna(0)
    lastdf['fill_qty'] = sumdf['fill_qty']
    lastdf['taker_fill_qty'] = sumdf['taker_fill_qty']
    lastdf['cancelled_qty'] = sumdf['cancelled_qty']
    df = lastdf.reset_index()
    df['order_submitted_time'] = df['order_submitted_time'].astype(float)
    chosen = ~numpy.isnan(df['order_submitted_time'])
    df = df.loc[chosen]

    # (X) order.submit_time < book.time < order.accept_time
    # (O) order.submit_time < book.time < order.accept_time + 10 secs (10*1e9 ns)

    order = None
    feed = None

    bbodict = {}

    for prod, prodbbos in bbos.items():
      dfbids = pd.DataFrame(prodbbos.bbo_bids)
      dfasks = pd.DataFrame(prodbbos.bbo_asks)
      bbodict[prod] = (dfbids, dfasks)

    for dfsymbol, orderdf in df.groupby('symbol'):
      if 'fill_price' not in orderdf.columns:
        continue
      orderdf = orderdf.reset_index(drop=True)
      product = order_symbol_to_product_map[dfsymbol]

      dfbids, dfasks = bbodict[product]
      if len(dfbids) == 0:
        print(f"skip {product}, empty feed")
        continue

      orderdf['sign'] = orderdf.order_side.astype(str).str.startswith("BUY").astype(
          int) - orderdf.order_side.astype(str).str.startswith("SELL").astype(int)
      for fwd_ms in [0, -10000, -3000, -1000,
                     25, 50, 100, 200, 500, 1000, 3000, 5000, 10000, 30000, 60000]:
        fwd_str = ('%d' % fwd_ms).replace("-", "m")
        for col, dfbbo in [("bid", dfbids), ("ask", dfasks)]:
          idxproj12 = bisect_batch(
              (orderdf['order_submitted_time'].astype(int) + fwd_ms * 1e6).tolist(),
              dfbbo['timestamp'].astype(int).tolist())
          idxproj12 = numpy.array(idxproj12)
          timediff = (
              orderdf['order_submitted_time'] + fwd_ms * 1e6
              - dfbbo['timestamp'].iloc[idxproj12].to_numpy())
          nan_idx = (idxproj12 < 0) | (timediff > 1e10) | (timediff < 0)
          fwd_str = ('%d' % fwd_ms).replace("-", "m")
          colname = f'{col}0_price_%sms_fwd' % fwd_str
          orderdf.loc[:, colname] = dfbbo['price'].iloc[idxproj12].reset_index(drop=True)
          orderdf.loc[nan_idx, colname] = numpy.nan
        assert (
            (orderdf['ask0_price_0ms_fwd'] - orderdf['bid0_price_0ms_fwd']).fillna(0) >= 0).all()
        orderdf['agg_price_%sms_fwd' % fwd_str] = \
            orderdf.loc[:, 'ask0_price_%sms_fwd' % fwd_str] * (orderdf['sign'] == 1) \
            + orderdf.loc[:, 'bid0_price_%sms_fwd' % fwd_str] * (orderdf['sign'] == -1)
        orderdf['mid_price_%sms_fwd' % fwd_str] = 0.5 * (
            orderdf.loc[:, 'ask0_price_%sms_fwd' % fwd_str]
            + orderdf.loc[:, 'bid0_price_%sms_fwd' % fwd_str])
        orderdf['order_payup_bps_%sms_fwd' % fwd_str] = (
            (orderdf['order_price'] - orderdf['agg_price_%sms_fwd' % fwd_str]) *
            (orderdf['sign'] == 1)) / orderdf['order_price'] * 1e4 + (
                (-orderdf['order_price'] + orderdf['agg_price_%sms_fwd' % fwd_str]) *
                (orderdf['sign'] == -1)) / orderdf['order_price'] * 1e4
        orderdf['fill_ret_bps_%sms_fwd' % fwd_str] = (
            (orderdf['fill_price'] - orderdf['mid_price_%sms_fwd' % fwd_str]) *
            (orderdf['sign'] == 1)) / orderdf['fill_price'] * 1e4 + (
                (-orderdf['fill_price'] + orderdf['mid_price_%sms_fwd' % fwd_str]) *
                (orderdf['sign'] == -1)) / orderdf['fill_price'] * 1e4
        orderdf['mid_ret_bps_%sms_fwd' % fwd_str] = (
            (orderdf['mid_price_0ms_fwd'] - orderdf['mid_price_%sms_fwd' % fwd_str]) *
            (orderdf['sign'] == 1)) / orderdf['mid_price_0ms_fwd'] * 1e4 + (
                (-orderdf['mid_price_0ms_fwd'] + orderdf['mid_price_%sms_fwd' % fwd_str]) *
                (orderdf['sign'] == -1)) / orderdf['mid_price_0ms_fwd'] * 1e4
        orderdf['order_ret_bps_%sms_fwd' % fwd_str] = (
            (orderdf['order_price'] - orderdf['mid_price_%sms_fwd' % fwd_str]) *
            (orderdf['sign'] == 1)) / orderdf['order_price'] * 1e4 + (
                (-orderdf['order_price'] + orderdf['mid_price_%sms_fwd' % fwd_str]) *
                (orderdf['sign'] == -1)) / orderdf['order_price'] * 1e4
        orderdf['payup0_ret_bps_%sms_fwd' % fwd_str] = (
            (orderdf['agg_price_0ms_fwd'] - orderdf['mid_price_%sms_fwd' % fwd_str]) *
            (orderdf['sign'] == 1)) / orderdf['order_price'] * 1e4 + (
                (-orderdf['agg_price_0ms_fwd'] + orderdf['mid_price_%sms_fwd' % fwd_str]) *
                (orderdf['sign'] == -1)) / orderdf['order_price'] * 1e4
        notnan = ~numpy.isnan(orderdf['order_payup_bps_%sms_fwd' % fwd_str])
        orderdf = orderdf.loc[notnan, :].reset_index(drop=True)

      orderdf['lifetime_ms'] = 1e-6 * numpy.nan_to_num(orderdf['dead_time'].astype(float)
                                                       - orderdf['order_submitted_time'])
      orderdf['fill_qty'] = orderdf['fill_qty'].fillna(0)
      orderdf['taker_fill_qty'] = orderdf['taker_fill_qty'].fillna(0)
      orderdf['spread_bps'] = 1e4 * (
          orderdf['ask0_price_0ms_fwd'] - orderdf['bid0_price_0ms_fwd']
              ) / orderdf['ask0_price_0ms_fwd']
      try:
        assert (orderdf['spread_bps'] >= 0).all()
      except:
        import pdb; pdb.set_trace()
      orderdf = orderdf.loc[orderdf['order_payup_bps_0ms_fwd'] >= 0, :]
      recorddf = orderdf.loc[:, Record._fields[7:]]
      recorddf['order_machine'] = row.machine
      recorddf['feed_machine'] = feed_machine
      recorddf['market_type'] = row.market_type
      recorddf['exchange'] = row.exchange
      relsymbol = product.subscription_symbol if hasattr(product,
                                                         "subscription_symbol") else product.symbol
      recorddf['symbol'] = relsymbol
      recorddf['timestamp'] = orderdf['order_submitted_time']

      if hasattr(product, "is_perpetual") and product.is_perpetual and row.exchange == 'Okex':
        recorddf['api_version'] = 'swap'
      else:
        recorddf['api_version'] = None

      recorddfs.append(recorddf.loc[:, Record._fields])
      if len(recorddfs) > 2:
        break
  return recorddfs


def run_day(date):
  dt = datetime.datetime.strptime(date, "%Y%m%d")
  start_ts = to_timestamp_int(dt)
  end_ts = to_timestamp_int(dt + datetime.timedelta(days=1))
  agg_types = [
    tss_pb2.STRATEGY_NAME,
    tss_pb2.STRATEGY_GROUP,
    tss_pb2.MACHINE,
    tss_pb2.MARKET_TYPE,
    tss_pb2.EXCHANGE,
    tss_pb2.OWNER,
    tss_pb2.SYMBOL,
  ]
  with XunkemgmtClient() as client:
    response = client.query_trading_summary_interval_histories(
      tss_pb2.QueryTradingSummaryIntervalHistoriesRequestProto(
        start_timestamp=start_ts,
        end_timestamp=end_ts,
        business_units=['coin', 'coinfof', 'prex', 'day1mm', 'extday1mm', 'digitalfund'],
        summary_info_only=True,
        agg_types=agg_types,
      )
    )
  data_list = []
  for proto in response.histories:
    summary_info = proto.summary_info
    data = {
      "machine": summary_info.machine,
      "market_type": summary_info.market_type,
      "exchange": summary_info.exchange,
      "owner": summary_info.owner,
      "strategy_group": summary_info.strategy_group,
      "strategy_name": summary_info.strategy_name,
      "symbol": summary_info.symbol,
    }
    data_list.append(data)
  df = pd.DataFrame(data_list)
  df.drop_duplicates(inplace=True)
  df.reset_index(drop=True, inplace=True)

  records = []

  jobs = []

  for (index, row) in df.iterrows():
    if flags.FLAGS.strat_name is not None and row.strategy_name != flags.FLAGS.strat_name:
      continue
    if flags.FLAGS.strat_group is not None and row.strategy_group != flags.FLAGS.strat_group:
      continue
    if flags.FLAGS.symbol is not None and row.symbol.find(flags.FLAGS.symbol) < 0:
      continue
    jobs.append((get_rows_records, ([row], date)))
    if os.environ.get("debug", None) is not None:
      get_rows_records([row], date)
  if os.environ.get("debug", None) is not None:
    return

  recorddfs = []

  from concurrent.futures import ProcessPoolExecutor, as_completed
  max_workers = 20
  with ProcessPoolExecutor(max_workers=max_workers) as executor:
    jobs_iter = iter(jobs)
    futures = {}

    while True:
      for job in jobs_iter:
        future = executor.submit(job[0], *job[1])
        futures[future] = job
        if len(futures) >= max_workers:
          break
      if len(futures) == 0:
        break

      for future in as_completed(futures):
        try:
          recorddfs.extend(future.result())
        except Exception:
          print(futures[future], "is broken. pass..")
        del futures[future]
        del future
        break

  if len(recorddfs) > 0:
    records = pd.concat(recorddfs, axis=0).reset_index(drop=True)
    print(f'converting {len(records)} records to DataFrame ..')
    print('saving ..')
    os.makedirs(f'{OUT_DIR}/dataframes/', exist_ok=True)
    records.to_hdf(f'{OUT_DIR}/dataframes/taker_ratio_{flags.FLAGS.prefix}{date}.h5', 'df')


def main(_):
  trading_dates = abase.get_trading_dates(flags.FLAGS.date)
  for tdate in trading_dates:
    run_day(tdate.strftime("%Y%m%d"))


if __name__ == "__main__":
  app.run(main)
